package com.cloud.service;

import org.apache.commons.lang3.ObjectUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;

/**
 * HBase 数据库操作工具
 *
 * @author yzj
 */
@Component
public class HBaseService {

    private static Logger logger = LoggerFactory.getLogger(HBaseService.class);

    @Resource
    private Admin hBaseAdmin;

    /**
     * 判断表是否存在
     *
     * @param tableName 表名
     * @return true/false
     */
    public boolean isExists(String tableName) {
        boolean tableExists = false;
        try {
            TableName table = TableName.valueOf(tableName);
            tableExists = hBaseAdmin.tableExists(table);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return tableExists;
    }

    /**
     * 创建表
     *
     * @param tableName    表名
     * @param columnFamily 列族
     * @return true/false
     */
    public boolean createTable(String tableName, List<String> columnFamily) {
        return createTable(tableName, columnFamily, null);
    }

    /**
     * 预分区创建表
     *
     * @param tableName  表名
     * @param familyList 列族
     * @param keys       分区集合
     * @return true/false
     */
    public boolean createTable(String tableName, List<String> familyList, List<String> keys) {
        if (!isExists(tableName)) {
            try {
                TableName table = TableName.valueOf(tableName);
                // 表建造者
                TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(table);
                // 列族建造者
                ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder;
                // 表描述器
                TableDescriptor tableDescriptor;
                ColumnFamilyDescriptor columnFamilyDescriptor;
                for (String family : familyList) {
                    columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family));
                    columnFamilyDescriptorBuilder.setMaxVersions(3);
                    columnFamilyDescriptor = columnFamilyDescriptorBuilder.build();
                    tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
                }
                tableDescriptor = tableDescriptorBuilder.build();
                if (ObjectUtils.isEmpty(keys)) {
                    hBaseAdmin.createTable(tableDescriptor);
                } else {
                    byte[][] splitKeys = getSplitKeys(keys);
                    hBaseAdmin.createTable(tableDescriptor, splitKeys);
                }
                return true;
            } catch (IOException e) {
                logger.error("HBase创建表" + tableName + "失败:" + e.getMessage());
            }
        } else {
            logger.info("HBase创建:" + tableName + ",表已存在!");
            return false;
        }
        return false;
    }

    /**
     * 删除表
     *
     * @param tableName 表名
     */
    public void dropTable(String tableName) {
        try {
            if (isExists(tableName)) {
                TableName table = TableName.valueOf(tableName);
                hBaseAdmin.disableTable(table);
                hBaseAdmin.deleteTable(table);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 插入数据（单条）
     *
     * @param tableName 表名
     * @param rowKey    rowKey
     * @param family    列族
     * @param column    列
     * @param value     值
     * @return true/false
     */
    public boolean saveData(String tableName, String rowKey, String family, String column, String value) {
        try {
            Table table = hBaseAdmin.getConnection().getTable(TableName.valueOf(tableName));
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column), Bytes.toBytes(value));
            table.put(put);
            table.close();
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 插入数据（批量）
     *
     * @param tableName 表名
     * @param rowKey    rowKey
     * @param family    列族
     * @param columns   列
     * @param values    值
     * @return true/false
     */
    public boolean batchData(String tableName, String rowKey, String family,
                             List<String> columns, List<String> values) {
        try {
            Table table = hBaseAdmin.getConnection().getTable(TableName.valueOf(tableName));
            Put put = new Put(Bytes.toBytes(rowKey));
            for (int i = 0; i < columns.size(); i++) {
                byte[] qualifier = Bytes.toBytes(columns.get(i));
                byte[] value = Bytes.toBytes(values.get(i));
                put.addColumn(Bytes.toBytes(family), qualifier, value);
            }
            table.put(put);
            table.close();
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 根据RowKey获取数据
     *
     * @param tableName 表名
     * @param rowKey    RowKey
     * @return map
     */
    public Map<String, String> getData(String tableName, String rowKey) {
        HashMap<String, String> map = new HashMap<>();
        try {
            Table table = hBaseAdmin.getConnection().getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);
            if (result != null && !result.isEmpty()) {
                for (Cell cell : result.listCells()) {
                    String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(),
                            cell.getFamilyLength()); // 列族
                    String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(),
                            cell.getQualifierLength()); // 列
                    String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(),
                            cell.getValueLength()); // 值
                    map.put(family + ":" + qualifier, value);
                }
            }
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return map;
    }

    /**
     * 获取数据（根据RowKey,列族,列）
     *
     * @param tableName 表名
     * @param rowKey    RowKey
     * @param family    列族
     * @param qualifier 列
     * @return map
     */
    public String getData(String tableName, String rowKey, String family, String qualifier) {
        String data = "";
        try {
            Table table = hBaseAdmin.getConnection().getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
            Result result = table.get(get);
            if (result != null && !result.isEmpty()) {
                Cell cell = result.listCells().get(0);
                data = Bytes.toString(cell.getValueArray(), cell.getValueOffset(),
                        cell.getValueLength());
            }
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return data;
    }

    /**
     * 删除数据（根据RowKey）
     *
     * @param tableName 表名
     * @param rowKey    RowKey
     */
    public void deleteData(String tableName, String rowKey) {
        try {
            Table table = hBaseAdmin.getConnection().getTable(TableName.valueOf(tableName));
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除数据（根据RowKey,列族）
     *
     * @param tableName 表名
     * @param rowKey    RowKey
     * @param family    列族
     */
    public void deleteData(String tableName, String rowKey, String family) {
        try {
            Table table = hBaseAdmin.getConnection().getTable(TableName.valueOf(tableName));
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            delete.addFamily(family.getBytes());
            table.delete(delete);
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除数据（根据RowKey，列族）
     *
     * @param tableName 表名
     * @param rowKey    RowKey
     * @param family    列族
     * @param column    列
     */
    public void deleteData(String tableName, String rowKey, String family, String column) {
        try {
            Table table = hBaseAdmin.getConnection().getTable(TableName.valueOf(tableName));
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            delete.addColumn(family.getBytes(), column.getBytes());
            table.delete(delete);
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除数据/多行
     *
     * @param tableName 表名
     * @param rowKeys   RowKey集合
     */
    public void deleteData(String tableName, List<String> rowKeys) {
        try {
            Table table = hBaseAdmin.getConnection().getTable(TableName.valueOf(tableName));
            List<Delete> deleteList = new ArrayList<>();
            for (String row : rowKeys) {
                Delete delete = new Delete(Bytes.toBytes(row));
                deleteList.add(delete);
            }
            table.delete(deleteList);
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 分区【10, 20, 30】 -> ( ,10] (10,20] (20,30] (30, )
     *
     * @param keys 分区集合[10, 20, 30]
     * @return byte二维数组
     */
    private byte[][] getSplitKeys(List<String> keys) {
        byte[][] splitKeys = new byte[keys.size()][];
        TreeSet<byte[]> rows = new TreeSet<>(Bytes.BYTES_COMPARATOR);
        for (String key : keys) {
            rows.add(Bytes.toBytes(key));
        }
        int i = 0;
        for (byte[] row : rows) {
            splitKeys[i] = row;
            i++;
        }
        return splitKeys;
    }

}
